package beautiful.butterfly.drds.binlog.consumer;

import beautiful.butterfly.drds.binlog.connection.ConnectionAttributes;
import beautiful.butterfly.drds.binlog.snapshot.SnapShot;
import beautiful.butterfly.drds.binlog.type.ParallelType;
import beautiful.butterfly.drds.binlog.utils.PropertiesExtend;
import beautiful.butterfly.drds.binlog.utils.StringUtils;
import beautiful.butterfly.drds.binlog.utils.TimeUtils;
import com.alibaba.fastjson.JSONObject;
import lombok.extern.slf4j.Slf4j;

import java.util.ArrayList;
import java.util.concurrent.atomic.AtomicLong;

@Slf4j
public class ConsumerMananger
{

    // 单例模式
    private static ConsumerMananger consumerMananger = null;

    static
    {
        consumerMananger = new ConsumerMananger(PropertiesExtend.getInstance().getConsumerWorker());
        new Thread(new Runnable()
        {
            // 打印剩余大小
            @Override
            public void run()
            {
                while (true)
                {
                    // 每6秒打印1次
                    TimeUtils.sleepMilliSeconds(6 * 1000);
                    // 搜集
                    String result = "[ totalItem-" + consumerMananger.getTotalNumber() + " ] [index - leftSize]";
                    Consumer[] consumers = consumerMananger.getConsumers();
                    for (Consumer consumer : consumers)
                    {
                        result += "[" + consumer.getIndex() + " - " + consumer.getQueueLeftSize() + "] ";
                    }
                    // 打印
                    log.info(result);
                }
            }
        }).start();
    }

    private Consumer[] consumers = null;
    private AtomicLong totalNumber = new AtomicLong(0);// 总的消费数量

    private ConsumerMananger(int count)
    {
        consumers = new Consumer[count];
        for (int index = 0; index < count; index++)
        {
            // 初始化+线程启动
            consumers[index] = new Consumer(index).start();
        }
    }

    public static ConsumerMananger getInstance()
    {
        return consumerMananger;
    }

    public static synchronized void start()
    {
        // 触发static块被执行
    }

    public void increase()
    {
        totalNumber.incrementAndGet();
    }

    public long getTotalNumber()
    {
        return totalNumber.get();
    }

    public Consumer[] getConsumers()
    {
        return consumers;
    }

    public void addSnapShot(SnapShot snapShot, ConnectionAttributes connectionAttributes)
    {
        JSONObject jsonObject;
        ArrayList<String> pks;
        String databaseName = snapShot.getDatabaseName();
        String tableName = snapShot.getTableName();
        String unionStr = StringUtils.union(databaseName, tableName);
        // 开始判断加速类型了
        ParallelType parallelType = snapShot.getParallelType();
        if (ParallelType.table == parallelType)
        {
            // 什么都不做
        } else
        {// 按行复制,shot里面肯定只有一条数据
            jsonObject = snapShot.getDatas().get(0);
            pks = snapShot.getPks();
            for (String pk : pks)
            {// 因为主键不一定是string类型
                unionStr = StringUtils.union(unionStr, jsonObject.get(pk).toString());
            }
        }
        int hashIndex = unionStr.hashCode() % consumers.length;
        if (hashIndex < 0)
        {
            hashIndex += consumers.length;
        }
        // LoggerUtils.debug(logger, "databaseName:tableName:" + database + ":" + tableName);
        // LoggerUtils.debug(logger,
        // "unionStr.hashCode():"+unionStr.hashCode());
        // LoggerUtils.debug(logger, "consumer index:"+hashIndex);
        Consumer consumer = consumers[hashIndex];
        consumer.addSnapShot(snapShot, connectionAttributes);
    }

}
